package com.winit.schedule.task.tracker.monitor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.conf.JobConfiguration;
import com.winit.schedule.core.protocol.command.HeartbeatRequest;
import com.winit.schedule.core.repository.entity.TaskRunnerEntity;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.service.AbstractService;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.task.tracker.TaskTracker;
import com.winit.schedule.task.tracker.TaskTrackerContextImpl;
import com.winit.schedule.task.tracker.conf.TaskConfiguration;
import com.winit.schedule.task.tracker.remoting.RemotingClientService;
import com.winit.schedule.task.tracker.runner.AbstractTaskManager;
import com.winit.schedule.task.tracker.runner.StrategyTask;
import com.winit.schedule.task.tracker.utils.DataConverter;
import com.winit.schedule.task.tracker.utils.RemoteUtil;

/**
 * 心跳监控
 * 
 * @author jianke.zhang 2015-3-27 下午12:14:18
 * @since 1.0
 */
public class HeartbeatMonitor extends AbstractService {

    private static transient final Logger  logger            = LoggerFactory.getLogger(HeartbeatMonitor.class);

    // 用来定时发送心跳
    private final ScheduledExecutorService heartbeatExecutor = Executors.newScheduledThreadPool(1);

    private long                           timerInterval;

    private long                           heartbeatTimeout;

    private RemotingClientService          remotingClientService;

    private TaskTracker                    taskTracker;

    private TaskTrackerContextImpl         taskTrackerContext;

    public HeartbeatMonitor(TaskTrackerContextImpl taskTrackerContext, RemotingClientService remotingClientService){
        super("HeartbeatMonitor");
        this.taskTrackerContext = taskTrackerContext;
        this.taskTracker = taskTrackerContext.getTaskTracker();
        this.remotingClientService = remotingClientService;
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);
        timerInterval = conf.getLong(TaskConfiguration.HEARTBEAT_INTERVAL_MS,
            TaskConfiguration.DEFAULT_HEARTBEAT_INTERVAL_MS);
        heartbeatTimeout = conf.getLong(TaskConfiguration.HEARTBEAT_TIMEOUT_MS,
            TaskConfiguration.DEFAULT_HEARTBEAT_TIMEOUT_MS);
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        heartbeatExecutor.scheduleWithFixedDelay(new HeartbeatThread(this), 2000, timerInterval, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
        taskTracker.stopAllTask();
        heartbeatExecutor.shutdown();
    }

    public TaskTracker getTaskTracker() {
        return taskTracker;
    }

    public TaskTrackerContextImpl getTaskTrackerContext() {
        return taskTrackerContext;
    }

    private List<TaskRunnerEntity> buildTaskRunners() {
        Collection<List<StrategyTask>> managerList = taskTracker.getManagerMap().values();

        List<TaskRunnerEntity> taskRunnerEntities = new ArrayList<TaskRunnerEntity>();
        TaskRunnerEntity taskRunnerEntity = null;
        for (List<StrategyTask> taskMgrs : managerList) {
            if (null != taskMgrs) {
                StrategyTask[] taskManagers = taskMgrs.toArray(new StrategyTask[] {});
                for (StrategyTask taskMgr : taskManagers) {
                    AbstractTaskManager taskManager = (AbstractTaskManager) taskMgr;
                    taskRunnerEntity = DataConverter.convert2TaskTunner(taskManager);
                    taskRunnerEntities.add(taskRunnerEntity);
                }
            }
        }
        return taskRunnerEntities;
    }

    class HeartbeatThread implements Runnable {

        // 连续心跳失败次数
        int              failureCount;

        boolean          reRegisterTask   = true;

        /**
         * 每次重新连接JobTracker会尝试对任务定义进行5次注册
         */
        int              registerTryCount = 0;

        HeartbeatMonitor heartbeatMonitor;

        public HeartbeatThread(HeartbeatMonitor heartbeatMonitor){
            this.heartbeatMonitor = heartbeatMonitor;
        }

        private HeartbeatRequest getHeartBeatRequest() throws SchedulerException {
            TaskTrackerEntity taskTrackerEntity = heartbeatMonitor.getTaskTracker().getTaskTrackerEntity();
            taskTrackerEntity.setHeartbeatTime(Clock.getSystemTime());
            TaskTrackerEntity newTaskTrackerEntity = new TaskTrackerEntity();
            BeanUtils.copyProperties(taskTrackerEntity, newTaskTrackerEntity);
            
            // 发送TaskTracker RPC调用超时时间给JobTracker
            newTaskTrackerEntity.setRpcTimeout(getConfig().getLong(JobConfiguration.CONNECT_TIMEOUT, JobConfiguration.DEFAULT_CONNECT_TIMEOUT));
            // 系统休眠时，会导致无法注册成功
//            if (!reRegisterTask) {
//                newTaskTrackerEntity.setTaskDefines(null);
//            }
            HeartbeatRequest body = new HeartbeatRequest(newTaskTrackerEntity, heartbeatMonitor.buildTaskRunners());
            
            if (null != taskTrackerContext.getQuartzScheduleService()) {
                body.setSchedulerEntities(taskTrackerContext.getQuartzScheduleService().getSchedulerJobs());
            }
            return body;
        }

        @Override
        public void run() {
            try {
                Thread.currentThread().setPriority(Thread.MAX_PRIORITY);

                if (remotingClientService.isStart()) {
                    boolean success = RemoteUtil.heartbeat(getHeartBeatRequest(), remotingClientService, taskTracker);
                    if (!success) {

                        logger.warn("heartbeat failure.");

                        if ((failureCount + 1) * timerInterval >= heartbeatTimeout) {
                            // 连续心跳失败的总时间大于配置的超时时间，停止所有作业;
                            if (!heartbeatMonitor.getTaskTracker().getManagerMap().isEmpty()) {
                                taskTracker.stopAllTask();
                                reRegisterTask = true;
                                registerTryCount = 0;
                            }
                        } else {
                            failureCount = failureCount + 1;
                        }
                        
                        if (failureCount > 0) {
                            logger.warn("Heartbeat failure times : " + failureCount);
                        }
                    } else {
                        if (reRegisterTask && ++registerTryCount > 5) {
                            reRegisterTask = false;
                            registerTryCount = 0;
                        }
                        failureCount = 0;
                    }

                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("waiting remotingClientService start.");
                    }
                }
            } catch (Exception ex) {
                logger.error(ex.getMessage(), ex);
            }
        }
    }

}
